package com.google.code.stalkernet.peer;

import java.io.DataOutput;
import java.io.DataOutputStream;
import java.io.IOException;
import java.io.OutputStream;
import java.nio.ByteBuffer;
import java.util.logging.Level;
import java.util.logging.Logger;

import com.google.code.stalkernet.async.Async.Callback;
import com.google.code.stalkernet.async.io.AsyncSocketChannel;
import com.google.code.stalkernet.message.Message;
import com.google.code.stalkernet.message.MessageUnmarshaller;
import com.google.code.stalkernet.message.MessageUtils;
import com.google.code.stalkernet.node.NodeID;

public class AsyncSocketChannelPeer extends AbstractPeer
{
    private final Logger logger = Logger.getLogger (
        AsyncSocketChannelPeer.class.getName ());

    private final ByteBuffer readBuffer = ByteBuffer.allocate (65536);
    private final MessageUnmarshaller messageUnmarshaller =
        new MessageUnmarshaller ();

    private final ByteBuffer writeBuffer = ByteBuffer.allocate (1024 * 1024);
    private final DataOutput dataOutput = new DataOutputStream (
        new ByteBufferOutputStream (writeBuffer));

    private final NodeID remoteNodeID;
    private final AsyncSocketChannel asyncSocketChannel;

    private final ReadCallback readCallback = new ReadCallback ();
    private final WriteCallback writeCallback = new WriteCallback ();

    private boolean writing = false;

    public AsyncSocketChannelPeer (
        NodeID remoteNodeID, AsyncSocketChannel asyncSocketChannel)
    {
        if (remoteNodeID == null)
            throw new IllegalArgumentException ("Remote node ID is null");

        if (asyncSocketChannel == null)
            throw new IllegalArgumentException ("Async socket channel is null");

        this.remoteNodeID = remoteNodeID;
        this.asyncSocketChannel = asyncSocketChannel;

        writeBuffer.flip ();

        asyncSocketChannel.read (readBuffer, readCallback);
    }

    public AsyncSocketChannel getAsyncSocketChannel ()
    {
        return asyncSocketChannel;
    }

    @Override
    public NodeID getRemoteNodeID ()
    {
        return remoteNodeID;
    }

    @Override
    public void sendMessage (Message message)
    {
        if (message == null)
            throw new IllegalArgumentException ("Message is null");

        logger.info ("Sending message: " + message);

        try
        {
            MessageUtils.marshalMessage (message, dataOutput);
        }
        catch (IOException ex)
        {
            throw new Error ("Impossible");
        }

        if (!writing)
        {
            writing = true;
            asyncSocketChannel.write (writeBuffer, writeCallback);
        }
    }

    @Override
    public void disconnect ()
    {
        try
        {
            asyncSocketChannel.shutdownOutput ();
        }
        catch (IOException ex)
        {
            logger.log (
                Level.SEVERE,
                "Cannot shutdown output on " + asyncSocketChannel,
                ex);
        }
        try
        {
            asyncSocketChannel.close ();
        }
        catch (IOException ex)
        {
            logger.log (
                Level.SEVERE,
                "Cannot close " + asyncSocketChannel,
                ex);
        }

        fireDisconnected ();
    }

    private class ReadCallback implements Callback <Integer>
    {
        @Override
        public void onSuccess (Integer result)
        {
            if (result == null)
                throw new IllegalArgumentException ("Result is null");

            int l = result.intValue ();

            if (l < 0)
            {
                logger.info ("Connection terminated: " + asyncSocketChannel);

                disconnect ();
            }
            else if (l > 0)
            {
                for (Message message:
                    messageUnmarshaller.unmarshall (
                        readBuffer.array (), 0, readBuffer.position ()))
                {
                    logger.info ("Message received: " + message);

                    fireMessageReceived (message);
                }

                readBuffer.clear ();
            }

            asyncSocketChannel.read (readBuffer, readCallback);
        }

        @Override
        public void onError (Throwable error)
        {
            logger.log (
                Level.SEVERE,
                "Cannot read from async socket channel: " + asyncSocketChannel,
                error);

            disconnect ();
        }
    }

    private class WriteCallback implements Callback <Integer>
    {

        @Override
        public void onSuccess (Integer result)
        {
            if (result == null)
                throw new IllegalArgumentException ("Result is null");

            writing = false;

            if (writeBuffer.remaining () > 0)
            {
                writing = true;
                asyncSocketChannel.write (writeBuffer, writeCallback);
            }
        }

        @Override
        public void onError (Throwable error)
        {
            writing = false;

            logger.log (
                Level.SEVERE,
                "Cannot write to async socket channel: " + asyncSocketChannel,
                error);

            disconnect ();
        }
    }

    private static class ByteBufferOutputStream extends OutputStream
    {
        private final ByteBuffer byteBuffer;

        public ByteBufferOutputStream (ByteBuffer byteBuffer)
        {
            if (byteBuffer == null)
                throw new IllegalArgumentException ("Byte buffer is null");

            this.byteBuffer = byteBuffer;
        }

        @Override
        public void write (int b) throws IOException
        {
            byteBuffer.compact ();
            byteBuffer.put ((byte)b);
            byteBuffer.flip ();
        }

        @Override
        public void write (byte [] b) throws IOException
        {
            if (b == null)
                throw new IllegalArgumentException ("B is null");

            byteBuffer.compact ();
            byteBuffer.put (b);
            byteBuffer.flip ();
        }

        @Override
        public void write (byte [] b, int off, int len) throws IOException
        {
            if (b == null)
                throw new IllegalArgumentException ("B is null");

            if (off < 0)
                throw new IllegalArgumentException ("Off (" + off + ") < 0");

            if (len < 0)
                throw new IllegalArgumentException ("Len (" + len + ") < 0");

            int bLength = b.length;

            if (off + len > bLength)
                throw new IllegalArgumentException ("Off (" + off +
                    ") + len (" + len + ") > b length (" + bLength + ")");

            byteBuffer.compact ();
            byteBuffer.put (b, off, len);
            byteBuffer.flip ();
        }
    }
}
